package com.freez.spark.core.demo;

import com.clearspring.analytics.util.Lists;
import com.freez.spark.tool.TupleUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.Optional;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.broadcast.Broadcast;
import scala.Tuple2;

import java.util.*;

/**
 * FREEDOM  2021 人生苦短，不妨一试
 * FileName: Markov.java
 * Author: zcs
 * Date: 2021年-12月-07日 周二 18:19
 * Description: TODO
 */
public class Markov {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setAppName("SparkDemo-markov").setMaster("local");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        Map<String, String> status = new HashMap<>();
        status.put("SL", "0");
        status.put("SE", "1");
        status.put("SG", "2");
        status.put("ML", "3");
        status.put("ME", "4");
        status.put("MG", "5");
        status.put("LL", "6");
        status.put("LE", "7");
        status.put("LG", "8");
        //广播
        Broadcast<Map<String, String>> mapBroadcast = jsc.broadcast(status);
        //初始化矩阵状态
        List<Tuple2<String, List<Double>>> statusList = initStatus(status.size());
        Broadcast<List<Tuple2<String, List<Double>>>> broadcastInitList = jsc.broadcast(statusList);


        markov(jsc, mapBroadcast, broadcastInitList);
    }

    public static void markov(JavaSparkContext jsc, Broadcast<Map<String, String>> mapBroadcast, Broadcast<List<Tuple2<String, List<Double>>>> broadcastInitList) {
        JavaRDD<String> lineRDD = jsc.textFile("./src/main/resources/data/core/demo/markov.txt");
        //RDD ===> (user,(2018-8-1,66))
        JavaPairRDD<String, Tuple2<Long, Integer>> pairRDD = lineRDD.mapToPair(s -> {
            String[] split = StringUtils.split(s, ",");
            String key = "user";
            long date = DateUtils.parseDate(split[0], "yyyy-MM-dd").getTime();
            Integer amount = Integer.parseInt(split[1]);
            Tuple2<Long, Integer> value = TupleUtils.T2(date, amount);
            Tuple2<String, Tuple2<Long, Integer>> result = TupleUtils.T2(key, value);
            return result;
        });
        JavaPairRDD<String, Iterable<Tuple2<Long, Integer>>> customerRDD = pairRDD.groupByKey();
        //创建状态序列
        JavaPairRDD<String, List<String>> statusSequence = customerRDD.mapValues(s -> {
            List<Tuple2<Long, Integer>> dateAmountList = Lists.newArrayList(s);
            // 对dateAmountList按照日期排序;
            Collections.sort(dateAmountList, (t1, t2) -> t1._1.compareTo(t2._1));
            return toStatusSequence(dateAmountList);
        });
        JavaPairRDD<Tuple2<String, String>, Integer> mode1 = statusSequence.flatMapToPair(s -> {
            List<String> seqStatus = s._2;
            //判断不能少于两条记录
            if (seqStatus == null || seqStatus.size() < 2) {
                return Collections.emptyIterator();
            }
            List<Tuple2<Tuple2<String, String>, Integer>> resultList = new ArrayList<>();
            for (int i = 0; i < seqStatus.size() - 1; i++) {
                String startStatus = seqStatus.get(i);
                String endStatus = seqStatus.get(i + 1);
                Tuple2<String, String> result = new Tuple2<>(startStatus, endStatus);
                resultList.add(TupleUtils.T2(result, 1));
            }
            return resultList.iterator();
        });
        //统计所有状态频率
        JavaPairRDD<Tuple2<String, String>, Integer> status1 = mode1.reduceByKey((v1, v2) -> (v1 + v2));
        JavaPairRDD<String, Tuple2<String, Integer>> status2 = status1.mapToPair(s -> {
            String key = s._1._1;
            Tuple2<String, Integer> value = TupleUtils.T2(s._1._2, s._2);
            return TupleUtils.T2(key, value);
        });
        JavaPairRDD<String, Iterable<Tuple2<String, Integer>>> groupStatus = status2.groupByKey().mapToPair(s -> {
            String rowNumber = mapBroadcast.getValue().get(s._1);
            return TupleUtils.T2(rowNumber, s._2);
        });
        JavaPairRDD<String, List<Double>> initRDD = jsc.parallelizePairs(broadcastInitList.getValue());
        JavaPairRDD<String, Tuple2<List<Double>, Optional<Iterable<Tuple2<String, Integer>>>>> leftJoinRDD = initRDD.leftOuterJoin(groupStatus);
        //规范化转移矩阵，是能行的概率和为1
        JavaPairRDD<String, List<Double>> resultRDD = leftJoinRDD.mapValues(new Function<Tuple2<List<Double>, Optional<Iterable<Tuple2<String, Integer>>>>, List<Double>>() {
            @Override
            public List<Double> call(Tuple2<List<Double>, Optional<Iterable<Tuple2<String, Integer>>>> s) throws Exception {

                int size = mapBroadcast.getValue().size();
                List<Double> doubleList = s._1;
                Optional<Iterable<Tuple2<String, Integer>>> optional = s._2;
                //
                if (optional.isPresent()) {
                    Iterable<Tuple2<String, Integer>> frequency = optional.get();
                    List<Tuple2<String, Integer>> frequencyList = Lists.newArrayList(frequency);
                    int sum = 0;
                    for (Tuple2<String, Integer> item : frequencyList) {
                        sum += item._2;
                    }
                    //防治概率为0
                    if (frequencyList.size() < size) {
                        sum += size;
                        for (int i = 0; i < doubleList.size(); i++) {
                            doubleList.set(i, 1.0 / sum);
                        }
                    }
                    for (int i = 0; i < frequencyList.size(); i++) {
                        String statusNumber = mapBroadcast.getValue().get(frequencyList.get(i)._1);
                        double value = frequencyList.get(i)._2 / (double) sum;
                        doubleList.set(Integer.parseInt(statusNumber), value);
                    }
                    System.out.println(frequencyList.size());
                } else {
                    return doubleList;
                }
                return doubleList;
            }
        });
        //打印最终结果
        for (Tuple2<String, List<Double>> item : resultRDD.sortByKey().collect()) {
            StringBuilder stringBuilder = new StringBuilder();
            stringBuilder.append(item._1).append("，");
            for (int i = 0; i < item._2.size(); i++) {
                stringBuilder.append(item._2.get(i)).append("|");
            }
            stringBuilder.append(item._2.get(item._2.size() - 1));
            System.out.println(stringBuilder);
        }
        jsc.close();
    }

    /**
     * 初始化矩阵状态 value = 1.0 / size
     *
     * @param size zidingyixuliededaxiao
     * @return
     */
    public static List<Tuple2<String, List<Double>>> initStatus(int size) {
        List<Tuple2<String, List<Double>>> initStatusList = new ArrayList<>();
        for (int row = 0; row < size; row++) {
            List<Double> list = new ArrayList<>();
            for (int col = 0; col < size; col++) {
                list.add(1.0 / (double) size);
            }
            initStatusList.add(new Tuple2<>(String.valueOf(row), list));
        }
        return initStatusList;
    }

    /**
     * 转换状态序列
     *
     * @param list Tuple2<Long, Integer>类型
     * @return
     */
    public static List<String> toStatusSequence(List<Tuple2<Long, Integer>> list) {
        //至少存在2个交易历史记录
        if (list.size() < 2) {
            return null;
        }
        List<String> statusSequenceList = new ArrayList<>();
        Tuple2<Long, Integer> before = list.get(0);
        for (int i = 1; i < list.size(); i++) {
            Tuple2<Long, Integer> current = list.get(i);
            Long beforeDate = before._1;
            Long currentDate = current._1;
            int beforeAmount = before._2;
            int currentAmount = current._2;
            long spanDate = (currentDate - beforeDate) / 24 * 60 * 60 * 100;
            int spanAmount = currentAmount - beforeAmount;
            //划分时间间隔
            String sD = "";
            if (spanDate < 30) {
                //时间间隔短
                sD = "S";
            } else if (spanDate < 60) {
                //时间间隔中等
                sD = "M";
            } else {
                //时间间隔长
                sD = "L";
            }
            //划分消费趋势
            String sA = "";
            if (spanAmount < 0.9 * beforeAmount) {
                //趋势变小
                sA = "L";
            } else if (spanAmount > 1.1 * beforeAmount) {
                //趋势平等
                sA = "E";
            } else {
                //趋势增大
                sA = "G";
            }
            String result = sD + sA;
            statusSequenceList.add(result);
            before = current;
        }
        return statusSequenceList;
    }
}